package yuekao9.dws;

import yuekao9.entity.*;
import java.time.Duration;
import java.util.TimeZone;
import com.alibaba.fastjson.JSON;
import java.text.SimpleDateFormat;
import org.apache.flink.util.Collector;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.java.tuple.*;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;

public class WideData {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

//        DataStreamSource<String> tuple4 = env.addSource(KafkaUtil.kafkaSource("dws_widedata"));
//        tuple4.print();
        DataStreamSource<String> tuple4 = env.socketTextStream("hadoop-single", 9999);
//        tuple4.print();
        SingleOutputStreamOperator<tm4_1> tm4_1data = tuple4.map(new MapFunction<String, tm4_1>() {
            @Override
            public tm4_1 map(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                String create_time = jsonObject.getJSONObject("f0").getString("create_time");
                String create_time_day = jsonObject.getJSONObject("f0").getString("create_time_day");
                Integer cargo_type = jsonObject.getJSONObject("f1").getInteger("cargo_type");
                String status = jsonObject.getJSONObject("f0").getString("status");
                String name = jsonObject.getJSONObject("f3").getString("name");
                String order_no = jsonObject.getJSONObject("f0").getString("order_no");
                Double amount = jsonObject.getJSONObject("f0").getDouble("amount");
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
                sdf.setTimeZone(TimeZone.getTimeZone("UTC+8"));
                long time = sdf.parse(jsonObject.getJSONObject("f0").getString("create_time")).getTime();
                return new tm4_1(create_time, create_time_day, cargo_type, status, name, order_no, amount, null, null, time);
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy
                .<tm4_1>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                .withTimestampAssigner((event, timestamp) -> event.getTs()));


//        tm4_1data.print();
        SingleOutputStreamOperator<Tuple4<String, String, Integer, Double>> process = tm4_1data.filter(x -> "60010".equals(x.getStatus())).keyBy(new KeySelector<tm4_1, Tuple2<String, String>>() {
                    @Override
                    public Tuple2<String, String> getKey(tm4_1 o) throws Exception {
                        return new Tuple2<>(o.getCreate_time_day(), o.getName());
                    }
                }).window(TumblingEventTimeWindows.of(Time.days(1)))
                .process(new ProcessWindowFunction<tm4_1, Tuple4<String, String, Integer, Double>, Tuple2<String, String>, TimeWindow>() {
                    @Override
                    public void process(Tuple2<String, String> stringStringTuple2, ProcessWindowFunction<tm4_1, Tuple4<String, String, Integer, Double>, Tuple2<String, String>, TimeWindow>.Context context, Iterable<tm4_1> iterable, Collector<Tuple4<String, String, Integer, Double>> collector) throws Exception {
                        String daytime = null;
                        Integer count = 0;
                        String statu = null;
                        Double sumamount = 0.0;
                        for (tm4_1 o : iterable) {
                            if ("60010".equals(o.getStatus())) {
                                count++;
                                daytime = o.getCreate_time_day();
                                statu = o.getStatus();
                                sumamount += o.getAmount();
                            }
                        }
                        collector.collect(new Tuple4<>(daytime, statu, count, sumamount));
                    }
                });

        process.print();


        env.execute();

    }
}
